package com.atguigu.chapter11;

import com.atguigu.chapter5.source.WaterSensor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @ClassName: Flink01_TableApi_BasicUse
 * @Description:
 * @Author: kele
 * @Date: 2021/4/12 18:40
 * <p>
 * 方式一：
 * 1、先创建stream环境，读取数据
 *      StreamExecutionEnvironment.getExecutionEnvironment(conf)，
 * 2、创建dynamic table环境，
 *      StreamTableEnvironment.create(流环境)，
 * 3、将流环境转化为动态表. 表的字段名从pojo的属性名自动抽取
 *      表环境.fromDataStream(Datastream);
 * 4、把动态表转换成流
 *      表环境.toAppendStream(table，row.class)  //将表中的每一行数据作为输出类型
 *
 * 流到表：
 *      表环境.formDataStream()
 * 表到流：
 *      表环境.toAppendStream()
 *      表环境.toRetractStream()
 *
 **/
public class Flink01_TableApi_BasicUse {

    public static void main(String[] args) throws Exception {

        //创建流的环境，根据流的环境创建table的环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        DataStreamSource<WaterSensor> ds = env.fromElements(
                new WaterSensor("sensor_1", 1000L, 10),
                new WaterSensor("sensor_1", 2000L, 20),
                new WaterSensor("sensor_2", 3000L, 30),
                new WaterSensor("sensor_1", 4000L, 40),
                new WaterSensor("sensor_1", 5000L, 50),
                new WaterSensor("sensor_2", 6000L, 60)
        );

        /**
         * 将流转化为动态表
         */
        Table table = tenv.fromDataStream(ds);

        //表的查询结果，还是一个表
        Table result = table
                .where($("id").isEqual("sensor_1"))
                /**
                 * 对表进行聚合操作
                 */
                .groupBy($("id"))
                .aggregate( $("vc").sum().as("vc_sum")  )
                .select($("id"), $("vc_sum"));

        /**
         * 动态表转化为流
         *
         * 普通的使用appen流
         * 涉及到修改数据的需要使用到撤回流
         */
       // DataStream<Row> stream = tenv.toAppendStream(result, Row.class);
        DataStream<Tuple2<Boolean, Row>> stream = tenv.toRetractStream(result, Row.class);


        stream.print();

        /**
         * 最后输出的是哪个，哪个就需要执行
         */
        env.execute();


    }

}
